package abacab.router;

import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import abacab.actor.Actor;
import abacab.exec.MessageExecutor;
import abacab.message.Envelope;

/**
 * A router that sends the message to the actor depending on a "key" found in the message
 * 
 * @param <K>
 *          the type of the key
 * @param <M>
 *          the type of handled messages
 */
public class KeyAwareRouter<K, M> extends AbstractRouter<M> {
	/** The actors */
	private final ConcurrentMap<K, Actor<? super M>>	actors;

	/** The key extractor */
	private final KeyExtractor<K, M>									keyExtractor;

	/**
	 * Create the router
	 * 
	 * @param keyExtractor
	 *          the key extractor
	 */
	public KeyAwareRouter(KeyExtractor<K, M> keyExtractor) {
		this.actors = new ConcurrentHashMap<K, Actor<? super M>>();
		this.keyExtractor = keyExtractor;
	}

	@Override
	@SuppressWarnings("unchecked")
	public Actor<? super M>[] getActors() {
		Collection<Actor<? super M>> a = this.actors.values();
		return a.toArray(new Actor[a.size()]);
	}

	@Override
	public void send(MessageExecutor executor, Envelope<M> message) {
		K key = this.keyExtractor.getKey(message.getMessage());
		Actor<? super M> actor = this.actors.get(key);
		if (actor == null) {
			actor = buildActor(key.toString());
			Actor<? super M> concActor = this.actors.putIfAbsent(key, actor);
			if (concActor != null) {
				actor = concActor;
			}
			else {
				actor.init();
			}
		}
		actor.getActorContext().getQueue().offer(message);
		executor.queueForExecution(actor);
	}

	/**
	 * Extract a key out of a message
	 * 
	 * @param <K>
	 *          the key
	 * @param <M>
	 *          the message
	 */
	public static interface KeyExtractor<K, M> {
		/**
		 * Get the key out of the message
		 * 
		 * @param message
		 *          the message
		 * @return the key
		 */
		K getKey(M message);
	}

	/**
	 * The default implementation that always returns the same key (0)
	 * 
	 * @param <M>
	 *          the message
	 */
	public static final class NoKey<M> implements KeyExtractor<Integer, M> {
		/** The key */
		private static final Integer	KEY	= Integer.valueOf(0);

		@Override
		public Integer getKey(M message) {
			return KEY;
		}
	}
}